package cn.com.tt.flink.sql.engine.pojo.sql;

import cn.com.bluemoon.bd.utils.RegexUtils;
import org.apache.commons.lang3.StringUtils;
import org.apache.flink.table.api.Schema;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.table.api.internal.TableEnvironmentImpl;
import org.apache.flink.table.catalog.CatalogTable;
import org.apache.flink.table.catalog.CatalogView;
import org.apache.flink.table.catalog.ObjectIdentifier;
import org.apache.flink.table.operations.Operation;
import org.apache.flink.table.operations.ddl.CreateTableOperation;
import org.apache.flink.table.operations.ddl.CreateViewOperation;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.List;
import java.util.stream.Collectors;

public class BmPrintSql extends AbstractBmFlinkSql {
    private static final Logger Logger = LoggerFactory.getLogger(BmPrintSql.class);
    private String printTxt;
    private String insertIntoSql;

    public static final String PLATFORM_PRINT_FIX = "platform_print_";

    public BmPrintSql(String printTxt, String sqlStatement, StreamTableEnvironment tableEnv) {
        super(sqlStatement, tableEnv);
        this.printTxt = printTxt;
    }

    @Override
    protected void innerProcess() {
        List<Operation> parse = ((TableEnvironmentImpl) tableEnv).getParser().parse(sqlStatement);
        Operation operation = parse.get(0);
        String tvName;
        Schema tableSchema;
        List<String> columnList;

        if (operation instanceof CreateViewOperation) {
            ObjectIdentifier viewIdentifier = ((CreateViewOperation) operation).getViewIdentifier();
            tvName = viewIdentifier.getObjectName();
            CatalogView catalogView = ((CreateViewOperation) operation).getCatalogView();
            tableSchema = catalogView.getUnresolvedSchema();
        } else if (operation instanceof CreateTableOperation && printSupportTable()) {
            ObjectIdentifier tableIdentifier = ((CreateTableOperation) operation).getTableIdentifier();
            tvName = tableIdentifier.getObjectName();
            CatalogTable catalogTable = ((CreateTableOperation) operation).getCatalogTable();
            tableSchema = catalogTable.getUnresolvedSchema();
        } else {
            Logger.info("打印功能不支持该类型的建表语句:" + sqlStatement);
            tableEnv.executeSql(sqlStatement);
            return;
        }

        columnList = tableSchema.getColumns().stream().map(Schema.UnresolvedColumn::getName).collect(Collectors.toList());
        String printDesc = getPrintDesc(tvName);
        String printViewName = PLATFORM_PRINT_FIX + tvName;
        String printTableSql = createPrintViewSql(printViewName, columnList);
        insertIntoSql = createInsertIntoViewSql(tvName, printViewName, printDesc, columnList);
        tableEnv.executeSql(sqlStatement);
        tableEnv.executeSql(printTableSql);
    }

    public boolean printSupportTable() {
        String s = RegexUtils.extractData(sqlStatement, "'connector'(.*?),");
        return s.contains("kafka");
    }

    private String getPrintDesc(String tvName) {
        String descPattern = "\\((.*?)\\)";
        String desc = RegexUtils.extractData(printTxt, descPattern);
        return StringUtils.isNotBlank(desc) ? desc.replace("'", "") : tvName;
    }

    protected String createPrintViewSql(String viewName, List<String> columnList) {
        String fields = columnList.stream().map(s -> "`" + s.replace("`", "") + "` string,\n").reduce("", (a, b) -> a + b);
        String fieldStr = "print_desc string,\n" + fields.substring(0, fields.lastIndexOf(","));
        return "create table " + viewName + "(\n" +
                fieldStr + "\n" +
                ")with(\n" +
                "'connector' = 'print'\n" +
                ")";
    }

    protected String createInsertIntoViewSql(String sourceView, String printView, String desc, List<String> columnList) {
        String fields = columnList.stream().map(s -> "coalesce('" + s + ":'||cast(`" + s + "` as string),'" + s + ":Null'),\n").reduce("", (a, b) -> a + b);
        String fieldStr = "'" + desc + "',\n" + fields.substring(0, fields.lastIndexOf(","));
        return "insert into " + printView + " select \n" +
                fieldStr + "\n" +
                "from " + sourceView;
    }

    public String getInsertIntSql() {
        return insertIntoSql;
    }

}
